package com.fangsheng.rocketmq;

import org.apache.rocketmq.client.consumer.DefaultLitePullConsumer;
import org.apache.rocketmq.client.consumer.DefaultMQPullConsumer;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.message.MessageQueue;
import org.apache.rocketmq.common.protocol.body.ClusterInfo;
import org.apache.rocketmq.common.protocol.body.TopicList;
import org.apache.rocketmq.common.protocol.route.BrokerData;
import org.apache.rocketmq.common.protocol.route.TopicRouteData;
import org.apache.rocketmq.tools.admin.DefaultMQAdminExt;
import org.junit.jupiter.api.Test;

import java.util.*;

/**
 * @创建人: 放生
 * @创建时间: 2022/1/19
 * @描述: 当前是 rocketmq 的简单使用 案列
 */
public class SimpleRocketMQ {


    @Test
    public void producer() throws Exception {
        DefaultMQProducer producer = new DefaultMQProducer("ooxx");
        producer.setNamesrvAddr("192.168.10.11:9876");
        producer.start();

        for (int i = 0; i < 10; i++) {
            Message message = new Message();
            message.setTopic("wula");//主题
            message.setTags("TagA");// 标签 ，用来区分同一个topic下的不同的业务标识数据
            message.setBody(("ooxx"+i).getBytes());//发送的消息数据
            message.setWaitStoreMsgOK(true); // 是否等待消息发送存储ok

//            // 方式一
//            SendResult sendResult = producer.send(message);//发送消息  这个是同步的发送，当前循环没发送一个会阻塞等待返回
//            System.out.println(sendResult);
//
//            // 方式二 异步发送
//            producer.send(message, new SendCallback() {
//                @Override
//                public void onSuccess(SendResult sendResult) {
//                    System.out.println("发送成功的回调");
//                }
//
//                @Override
//                public void onException(Throwable throwable) {
//                    System.out.println("发送异常的回调");
//                }
//            });
//            // 方式三   只管发送  不管有没有发送成功，也不等待
//            producer.sendOneway(message);

            //方式四   往指定的 topic brokerName 队列 发送数据
            MessageQueue messageQueue  =  new MessageQueue();
            messageQueue.setTopic("wula");
            messageQueue.setBrokerName("hadoop11");
            messageQueue.setQueueId(0);
            SendResult send = producer.send(message, messageQueue);
            System.out.println(send);

        }






    }



    @Test
    public void admin() throws Exception {

        DefaultMQAdminExt admin = new DefaultMQAdminExt();
        admin.setNamesrvAddr("192.168.10.11:9876");
        admin.start();

        // 获取到当前的nameServer 主题的列表
        TopicList topicList = admin.fetchAllTopicList();


        Set<String> topicList1 = topicList.getTopicList();
        topicList1.stream().forEach(System.out::println);
        System.out.println("------------------------------------");
        TopicRouteData wula = admin.examineTopicRouteInfo("wula");
        System.out.println(wula);

    }


    @Test
    public void consumerPush() throws Exception {
        // 同时创建绑定一个消费者组 consumer_ox
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("consumer_ox");
        // 设置的 nameserver
        consumer.setNamesrvAddr("192.168.10.11:9876");
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);

        consumer.subscribe("wula","*");

        consumer.registerMessageListener(new MessageListenerConcurrently() {
            //小批量，小集合
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {

                msgs.forEach(messageExt -> {
//                    System.out.println(messageExt);
                    byte[] body = messageExt.getBody();
                    String msg = new String(body);
                    System.out.println(msg);
                });

                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
        consumer.start();

        System.in.read();


    }

    @Test
    public void consumerPull() throws Exception {
        DefaultLitePullConsumer consumer = new DefaultLitePullConsumer("xxx1x");
        consumer.setNamesrvAddr("192.168.10.11:9876");
        consumer.start();

        Collection<MessageQueue> mqs = consumer.fetchMessageQueues("wula");

        System.out.println("queues:");
        mqs.forEach(messageQueue -> System.out.println(messageQueue));

        System.out.println("poll....");
//        consumer.assign(mqs);  // 方式一


        // 方式二  精准消费，根据主题，节点node，队列的ID
        Collection<MessageQueue> queue = new ArrayList<>();
        MessageQueue qu = new MessageQueue("wula", "hadoop11", 0);
        queue.add(qu);
        consumer.assign(queue);
        consumer.seek(qu,4); // 表示从qu的队列  偏移量为4 开始消费
        List<MessageExt> poll = consumer.poll();
        poll.forEach(s-> System.out.println(s));


        System.in.read();

    }





}
